package com.zh.hbase;

import java.io.IOException;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Mapper;


/**
 * HBase MapReduce
 * 
 * 读一张表，放入另一张表
 * 
 * ImmutableBytesWritable 对应的key，也就是hbase的rowkey,根据rowkey一个个的读。
 * 每一个mapper读取的是一个rowkey。每一次处理的是一个rowkey里的所有cell。
 * hbase:一个region对应一个maptask
 * 
 * Result >> 集合类，对应的是cell
 * 
 */
public class TestHBaseMapper extends TableMapper<ImmutableBytesWritable,Put>{
	
	@Override
	protected void map(ImmutableBytesWritable key, Result value,
			Mapper<ImmutableBytesWritable,Result,ImmutableBytesWritable,Put>.Context context)
			throws IOException, InterruptedException {
		//封装put对象。put中封装了rowkey的所有数据
		Put put = new Put(key.get());
		//遍历结果集
		for(Cell cell:value.rawCells()){
			//判断当前的列簇
			if("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))){
				//判断当前列是不是name
				if("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
					put.add(cell);
				}
			}
		}
		
		//map的输出
		context.write(key, put);
		
	}
	
	
}
